package SQL.scan;

import java.util.ArrayList;

import exceptions.DBQueryGroupbyException;
import exceptions.DBTypeNotMatchException;
import Common.Types;
import SQL.abysnSQL.ColName;
import SQL.abysnSQL.Func;
import SQL.abysnSQL.HvCondition;
import SQL.abysnSQL.SelectExpList;
import SQL.abysnSQL.SelectExpr;
import SQL.abysnSQL.Value;
import SQL.abysnSQL.Func.FUNC;
import SQL.abysnSQL.SelectExpr.TYPE;
import Transaction.Transaction;
import heap.ResultField;
import heap.ResultRow;
import heap.ResultSchema;

public class GroupByScan implements Scan {
//	new GroupByScan(p.open(),new ColName(colName),dbname,sellist,hving,tx);
	public GroupByScan(
			Scan s, 
			ColName colName,
			String db,
			SelectExpList fieldList, 
			HvCondition having,
			Transaction tx) 
	throws Exception {
		this.colName = colName.name;
		this.having = having;
		this.db = db;
		this.tx = tx;
		this.fields = fieldList;
		this.order = new ArrayList<Integer>();
		this.schema = new ResultSchema();
		ArrayList<SortKey> keys = new ArrayList<SortKey>();
		keys.add(new SortKey(colName.name, true));
		this.s = new SortScan(s, db, keys, tx);
		tSchema = s.getSchema();
		for (int i = 0; i < tSchema.size(); i++) {
			if (tSchema.get(i).name.endsWith("."+colName) ||
					tSchema.get(i).name.equals(colName)) {
				this.group = i;
				break;
			}
		}
		while (fieldList != null) {
			if (fieldList.selExpr.value != null) {
				Value v = fieldList.selExpr.value;
				int value = v.getType(tSchema, db, tx);
				String name = v.toString();
				schema.add(new ResultField(name, value));
				for (int i = 0; i < tSchema.size(); i++) {
					if (tSchema.get(i).name.endsWith("."+name) ||
							tSchema.get(i).name.equals(name)) {
						order.add(i);
						break;
					}
				}
			}
			else {
				String name = fieldList.selExpr.colname.toString();
				ResultField rf = null;
				for (int i = 0; i < tSchema.size(); i++) {
					if (tSchema.get(i).name.endsWith("."+name) ||
							tSchema.get(i).name.equals(name)) {
						order.add(i);
						break;
					}
				}
				switch (fieldList.selExpr.func.value) {
				case AVG:
					rf = new ResultField("AVG("+name+")", Types.FLOAT);
					break;
				case COUNT:
					rf = new ResultField("SUM("+name+")", Types.INT);
					break;
				case SUM:
					rf = new ResultField("SUM("+name+")", 0);
					for (int i = 0; i < tSchema.size(); i++) {
						if (tSchema.get(i).name.endsWith("."+name) ||
								tSchema.get(i).name.equals(name)) {
							rf.type = tSchema.get(i).type;
							break;
						}
					}
					break;
				case MIN:
					rf = new ResultField("MIN("+name+")", 0);
					for (int i = 0; i < tSchema.size(); i++) {
						if (tSchema.get(i).name.endsWith("."+name) ||
								tSchema.get(i).name.equals(name)) {
							rf.type = tSchema.get(i).type;
							break;
						}
					}
					break;
				case MAX:
					rf = new ResultField("MAX("+name+")", 0);
					for (int i = 0; i < tSchema.size(); i++) {
						if (tSchema.get(i).name.endsWith("."+name) ||
								tSchema.get(i).name.equals(name)) {
							rf.type = tSchema.get(i).type;
							break;
						}
					}
					break;
				}
				schema.add(rf);
			}
			fieldList = fieldList.next;
		}
		currentScanResult = this.s.next();
	}

	public void beforeFirst() throws Exception {
		s.beforeFirst();
		currentScanResult = s.next();
	}

	public ResultRow getCurrent() throws Exception {
		return currentResult;
	}

	public ResultSchema getSchema() throws Exception {
		return schema;
	}

	public ResultRow next() throws Exception {
		if (currentScanResult == null) 
			return null;
		ResultRow nextScanResult;
		int count;
		ArrayList<String> result;
		while (true) {
			nextScanResult = currentScanResult;
			count = 0;
			result = new ArrayList<String>();
			do {
				currentScanResult = nextScanResult;
				if (having != null)
					having.addTuple(currentScanResult, tSchema, db, tx);
//				nextScanResult = s.next();
				
				SelectExpList list = fields;
				int i = 0;
				count++;
				while (list != null) {
					if (list.selExpr.value != null) {
						if (result.size() == i) {
							result.add(currentScanResult.get(order.get(i)));
						}
					}
					else {
						switch(list.selExpr.func.value) {
						case COUNT:
							if (result.size() == i) {
								result.add("1");
							} else {
								int p = Integer.parseInt(result.get(i))+1;
								result.set(i, p+"");
							}
							break;
						case AVG:
							if (result.size() == i) {
								result.add(currentScanResult.get(order.get(i)));
							} else {
								float p = Float.parseFloat(result.get(i));
								p += Float.parseFloat(currentScanResult.get(order.get(i)));
								result.set(i, p+"");
							}
							break;
						case SUM:
							if (result.size() == i) {
								result.add(currentScanResult.get(order.get(i)));
							} else {
								float p = Float.parseFloat(result.get(i));
								p += Float.parseFloat(currentScanResult.get(order.get(i)));
								result.set(i, p+"");
							}
							break;
						case MAX:
							if (result.size() == i) {
								result.add(currentScanResult.get(order.get(i)));
							} else {
								switch(schema.get(i).type) {
								case Types.INT:
								case Types.BOOLEAN:
								case Types.FLOAT:
									float a = Float.parseFloat(result.get(i));
									float b = Float.parseFloat(currentScanResult.get(order.get(i)));
									if (a < b) {
										result.set(i, currentScanResult.get(order.get(i)));
									}
									break;
								default:
									String aa = result.get(i);
								String bb = currentScanResult.get(order.get(i));
								if (aa.compareTo(bb) < 0) {
									result.set(i, bb);
								}
								}
							}
							break;
						case MIN:
							if (result.size() == i) {
								result.add(currentScanResult.get(order.get(i)));
							} else {
								switch(schema.get(i).type) {
								case Types.INT:
								case Types.BOOLEAN:
								case Types.FLOAT:
									float a = Float.parseFloat(result.get(i));
									float b = Float.parseFloat(currentScanResult.get(order.get(i)));
									if (a > b) {
										result.set(i, currentScanResult.get(order.get(i)));
									}
									break;
								default:
									String aa = result.get(i);
								String bb = currentScanResult.get(order.get(i));
								if (aa.compareTo(bb) > 0) {
									result.set(i, bb);
								}
								}
							}
							break;
						}
					}
					list = list.next;
					i++;
				}
				nextScanResult = s.next();

			} while (nextScanResult != null && 
					currentScanResult.get(group).equals(nextScanResult.get(group)));
			currentScanResult = nextScanResult;
			if (having == null || having.verify(tSchema)) {
				break;
			} else if (currentScanResult == null) {
				return null;
			}
		}
		SelectExpList list = fields;
		int i = 0;
		currentResult = new ResultRow();
		while (list != null) {
			if (list.selExpr.value != null) {
				currentResult.add(result.get(i));
			} else {
				switch(list.selExpr.func.value) {
				case AVG:
					float p = Float.parseFloat(result.get(i));
					p = p / count;
					currentResult.add(p+"");
					break;
				case SUM:
					if(schema.get(i).type == Types.INT) {
						currentResult.add(result.get(i).split("\\.")[0]);
						break;
					}
				default:
					currentResult.add(result.get(i));
				}
			}
			list = list.next;
			i++;
		}
		return currentResult;
	}

	public void close() throws Exception {
		s.close();
	}
	/* (non-Javadoc)
	 * @see SQL.scan.Scan#hasField(java.lang.String)
	 */
	@Override
	public boolean hasField(String fldname) throws Exception {
		// TODO Auto-generated method stub
		return false;
	}

	private HvCondition having;
	private ResultRow currentScanResult;
	private ResultSchema schema;
	private ResultRow currentResult;
	private String colName;
	private ArrayList<Integer> order;
	private int group;
	private String db;
	private Transaction tx;
	private ResultSchema tSchema;
	private Scan s;
	private SelectExpList fields;


}
